package cn.controlsys.common.mq.builder;


import cn.controlsys.common.mq.MQConstants;
import cn.controlsys.common.mq.interfaces.MessageSender;
import cn.controlsys.common.mq.keyid.IdWorkerFactory;
import cn.controlsys.common.mq.object.DetailResponse;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
//import org.springframework.amqp.rabbit.connection.ConnectionFactory;
//import org.springframework.amqp.rabbit.core.RabbitTemplate;
//import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Author: shanzheng <br/>
 * Description: RabbitMQ构建者<br/>
 * Date： 14:22 2018/1/26.<br/>
 * Modified By: <br/>
 */
@Slf4j
public class RabbitMQAccessBuilder {
    private ConnectionFactory connectionFactory;

    public RabbitMQAccessBuilder(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    /**
     * Description: 忘交换器类型是direct上发消息
     *
     * @param exchange   交换器名
     * @param routingKey 路由键
     * @param queue      队列
     * @author: shanzheng
     * @date: 19:59 2018/4/26
     * @return: cn.controlsys.job.common.mq.interfaces.MessageSender
     */
    public MessageSender buildDirectMessageSenderAck(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSenderAck(exchange, routingKey, queue, MQConstants.EXCHANGE_DIRECT_TYPE);
    }

    /**
     * Description: 忘交换器类型是direct上发消息
     *
     * @param exchange   交换器名
     * @param routingKey 路由键
     * @param queue      队列
     * @author: shanzheng
     * @date: 19:59 2018/4/26
     * @return: cn.controlsys.job.common.mq.interfaces.MessageSender
     */
    public MessageSender buildDirectMessageSenderNoAck(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSenderNoAck(exchange, routingKey, queue, MQConstants.EXCHANGE_DIRECT_TYPE);
    }

    /**
     * Description: 往交换器类型是topic上发消息 -- 但是没有绑定队列,我觉得这个方法没用
     *
     * @param exchange
     * @param routingKey
     * @author: shanzheng
     * @date: 20:00 2018/4/26
     * @return: cn.controlsys.job.common.mq.interfaces.MessageSender
     */
    public MessageSender buildTopicMessageSenderAck(final String exchange, final String routingKey) throws IOException {
        return buildMessageSenderAck(exchange, routingKey, null, MQConstants.EXCHANGE_TOPIC_TYPE);
    }

    /**
     * Description: 往交换器类型是topic上发消息
     *
     * @param exchange   交换机
     * @param routingKey 路由
     * @param queue      队列
     * @author: shanzheng
     * @date: 15:55 2018/1/26
     * @return: cn.controlsys.job.client.mq.interfaces.MessageSender
     */
    public MessageSender buildTopicMessageSenderAck(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSenderAck(exchange, routingKey, queue, MQConstants.EXCHANGE_TOPIC_TYPE);
    }

    /**
     * Description: 往交换器类型是topic上发消息 -- 无确认机制
     *
     * @param exchange   交换机
     * @param routingKey 路由
     * @param queue      队列
     * @author: shanzheng
     * @date: 15:55 2018/1/26
     * @return: cn.controlsys.job.client.mq.interfaces.MessageSender
     */
    public MessageSender buildTopicMessageSenderNoAck(final String exchange, final String routingKey, final String queue) throws IOException {
        return buildMessageSenderNoAck(exchange, routingKey, queue, MQConstants.EXCHANGE_TOPIC_TYPE);
    }

    /**
     * Description: 绑定并发送消息，有确认机制
     *
     * @param exchange   交换器名
     * @param routingKey 路由键
     * @param queue      队列
     * @param type       交换器类型
     * @author: shanzheng
     * @date: 20:03 2018/4/26
     * @return: cn.controlsys.job.common.mq.interfaces.MessageSender
     */
    //1 构造template, exchange, routingkey等
    //2 设置message序列化方法
    //3 设置发送确认
    //4 构造sender方法
    public MessageSender buildMessageSenderAck(final String exchange, final String routingKey,
                                               final String queue, final String type) throws IOException {
        Connection connection = connectionFactory.createConnection();
        //1 将队列、交换机、路由选择绑定
        if (MQConstants.EXCHANGE_DIRECT_TYPE.equals(type)) {// type.equals("direct")
            buildQueue(exchange, routingKey, queue, connection, type);
        } else if (MQConstants.EXCHANGE_TOPIC_TYPE.equals(type)) {// type.equals("topic")
            if (StringUtils.isNotEmpty(queue)) { //这个判断条件是我加的
                buildQueue(exchange, routingKey, queue, connection, type);
            } else {
                buildTopic(exchange, connection);
            }
        }

        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setExchange(exchange);
        rabbitTemplate.setRoutingKey(routingKey);
        //2 设置message序列化方法
//        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
//        RetryCache retryCache = new RetryCache(); // 重试机制先关掉

        //3 设置发送确认 发消息时return在confirm之前，所以验证的时候要confirm在前
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.info("send message failed: " + cause + correlationData.toString());
            } else {
                log.info("消息消费成功");
//                retryCache.del(correlationData.getId());// 重试机制先关掉
            }
        });

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                System.out.println("什么时候走");
                Thread.sleep(MQConstants.ONE_MINUTE);// 这里为了测试将ONE_SECOND 变为 ONE_MINUTE
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            log.info("send message failed: " + replyCode + " " + replyText);
            rabbitTemplate.send(message);
        });

        //4
        return new MessageSender() {
            {
//                retryCache.setSender(this);// 重试机制先关掉
            }

            @Override
            public DetailResponse send(Object message) {
                String id;
                try {
//                    String id = retryCache.generateId(); // id生成策略换掉
                    id = IdWorkerFactory.getWorkerId();
//                    retryCache.add(id, message); // 重试先关掉
                    rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(id));
                } catch (Exception e) {
                    return new DetailResponse(false, "", "");
                }

                return new DetailResponse(true, "", id);
            }

            @Override
            public DetailResponse send(String messageId, byte[] message) {
                String id;
                try {
//                    retryCache.add(messageId, message);// 重试机制先关掉
                    rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(messageId));
                } catch (Exception e) {
                    return new DetailResponse(false, "byte[] fail", "");
                }
                return new DetailResponse(true, "", messageId);
            }

            @Override
            public DetailResponse send(byte[] message) {
                String id;
                try {
                    id = IdWorkerFactory.getWorkerId();
//                    retryCache.add(id, message);// 重试机制先关掉
                    rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(id));
                } catch (Exception e) {
                    return new DetailResponse(false, "byte[] fail", "");
                }
                return new DetailResponse(true, "", id);
            }

            @Override
            public RabbitTemplate getTemplate() {
                return rabbitTemplate;
            }
        };
    }

    /**
     * Description: 绑定并发送消息，无确认机制
     *
     * @param exchange   交换机
     * @param routingKey 路由键
     * @param queue      队列
     * @param type       交换机类型 fanout,
     * @author: shanzheng
     * @date: 17:30 2018/4/26
     * @return: cn.controlsys.job.common.mq.interfaces.MessageSender
     */
    /* 发送不需要确认的消息 start*/
    //1 构造template, exchange, routingkey等
    //2 设置message序列化方法
    //4 构造sender方法
    public MessageSender buildMessageSenderNoAck(final String exchange, final String routingKey,
                                                 final String queue, final String type) throws IOException {
        Connection connection = connectionFactory.createConnection();
        //1 将队列、交换机、路由选择绑定
        if (MQConstants.EXCHANGE_DIRECT_TYPE.equals(type)) {// type.equals("direct")
            buildQueue(exchange, routingKey, queue, connection, type);
        } else if (MQConstants.EXCHANGE_TOPIC_TYPE.equals(type)) {// type.equals("topic")
            if (StringUtils.isNotEmpty(queue)) { //这个判断条件是我加的
                buildQueue(exchange, routingKey, queue, connection, type);
            } else {
                buildTopic(exchange, connection);
            }
        }

        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

        rabbitTemplate.setExchange(exchange);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setRoutingKey(routingKey);
        //2 设置message序列化方法
//        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
//        RetryCache retryCache = new RetryCache(); // 重试机制先关掉

        //3 设置发送确认 发消息时return在confirm之前，所以验证的时候要confirm在前 ，判断是否发送到了服务端
//        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
//            if (!ack) {
//                log.info("send message failed: " + cause + correlationData.toString());
//            } else {
//                log.info("消息消费成功");
////                retryCache.del(correlationData.getId());// 重试机制先关掉
//            }
//        });

        rabbitTemplate.setReturnCallback((message, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                System.out.println("什么时候走");
                Thread.sleep(MQConstants.ONE_MINUTE);// 这里为了测试将ONE_SECOND 变为 ONE_MINUTE
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            log.info("send message failed: " + replyCode + " " + replyText);
            rabbitTemplate.send(message);// 没有发送到交换机是否再次发送
        });

        //4
        return new MessageSender() {
            {
//                retryCache.setSender(this);// 重试机制先关掉
            }

            @Override
            public DetailResponse send(Object message) {
                String id;
                try {
                    id = IdWorkerFactory.getWorkerId();
//                    retryCache.add(id, message); // 重试先关掉
                    rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(id));
                } catch (Exception e) {
                    return new DetailResponse(false, "", "");
                }

                return new DetailResponse(true, "", id);
            }

            @Override
            public DetailResponse send(String messageId, byte[] message) {
                String id;
                try {
//                    id = IdWorkerFactory.getWorkerId();
//                    retryCache.add(messageId, message);// 重试机制先关掉
                    rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(messageId));
                } catch (Exception e) {
                    return new DetailResponse(false, "byte[] fail", "");
                }
                return new DetailResponse(true, "", messageId);
            }

            @Override
            public DetailResponse send(byte[] message) {
                String id;
                try {
                    id = IdWorkerFactory.getWorkerId();
//                    retryCache.add(id, message);// 重试机制先关掉
                    rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(id));
                } catch (Exception e) {
                    return new DetailResponse(false, "byte[] fail", "");
                }
                return new DetailResponse(true, "", id);

            }
            @Override
            public RabbitTemplate getTemplate() {
                return rabbitTemplate;
            }
        };
    }
    /* 发送不需要确认的消息 end*/


//    public <T> MessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue,
//                                                    final MessageProcess<T> messageProcess) throws IOException {
//        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, "direct");
//    }

//    public <T> MessageConsumer buildTopicMessageConsumer(String exchange, String routingKey, final String queue,
//                                                         final MessageProcess<T> messageProcess) throws IOException {
//        return buildMessageConsumer(exchange, routingKey, queue, messageProcess, "topic");
//    }

    //1 创建连接和channel
    //2 设置message序列化方法
    //3 构造consumer
//    public <T> MessageConsumer buildMessageConsumer(String exchange, String routingKey, final String queue,
//                                                    final MessageProcess<T> messageProcess, String type) throws IOException {
//        final Connection connection = connectionFactory.createConnection();
//
//        //1
//        buildQueue(exchange, routingKey, queue, connection, type);
//
//        //2
//        final MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
//        final MessageConverter messageConverter = new Jackson2JsonMessageConverter();
//
//        //3
//        return new MessageConsumer() {
//            QueueingConsumer consumer;
//
//            {
//                consumer = buildQueueConsumer(connection, queue);
//            }
//
//            @Override
//            //1 通过delivery获取原始数据
//            //2 将原始数据转换为特定类型的包
//            //3 处理数据
//            //4 手动发送ack确认
//            public DetailResponse consume() {
//                QueueingConsumer.Delivery delivery;
//                Channel channel = consumer.getChannel();
//
//                try {
//                    //1
//                    delivery = consumer.nextDelivery();
//                    Message message = new Message(delivery.getBody(),
//                            messagePropertiesConverter.toMessageProperties(delivery.getProperties(), delivery.getEnvelope(), "UTF-8"));
//
//                    //2
//                    @SuppressWarnings("unchecked")
//                    T messageBean = (T) messageConverter.fromMessage(message);
//
//                    //3
//                    DetailResponse detailRes;
//
//                    try {
//                        detailRes = messageProcess.process(messageBean);
//                    } catch (Exception e) {
//                        detailRes = new DetailResponse(false, "process exception: " + e,"");
//                    }
//
//                    //4
//                    if (detailRes.isSuccess()) {
//                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
//                    } else {
//                        //避免过多失败log
//                        Thread.sleep(MQConstants.ONE_SECOND);
//                        log.info("process message failed: " + detailRes.getErrMsg());
//                        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
//                    }
//
//                    return detailRes;
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                    return new DetailResponse(false, "interrupted exception " + e.toString(),"");
//                } catch (ShutdownSignalException | ConsumerCancelledException | IOException e) {
//                    e.printStackTrace();
//
//                    try {
//                        channel.close();
//                    } catch (IOException | TimeoutException ex) {
//                        ex.printStackTrace();
//                    }
//
//                    consumer = buildQueueConsumer(connection, queue);
//
//                    return new DetailResponse(false, "shutdown or cancelled exception " + e.toString(),"");
//                } catch (Exception e) {
//                    e.printStackTrace();
//                    log.info("exception : ", e);
//
//                    try {
//                        channel.close();
//                    } catch (IOException | TimeoutException ex) {
//                        ex.printStackTrace();
//                    }
//
//                    consumer = buildQueueConsumer(connection, queue);
//
//                    return new DetailResponse(false, "exception " + e.toString(),"");
//                }
//            }
//        };
//    }

    private void buildQueue(String exchange, String routingKey,
                            final String queue, Connection connection, String type) throws IOException {
        Channel channel = connection.createChannel(false); // false代表不使用事务 true则相反
        if (MQConstants.EXCHANGE_DIRECT_TYPE.equals(type)) {
            // 交换机声明
            channel.exchangeDeclare(exchange, MQConstants.EXCHANGE_DIRECT_TYPE, true, false, null);
        } else if (MQConstants.EXCHANGE_TOPIC_TYPE.equals(type)) {//  type.equals("topic")
            channel.exchangeDeclare(exchange, MQConstants.EXCHANGE_TOPIC_TYPE, true, false, null);
        }
        // 队列声明
        channel.queueDeclare(queue, true, false, false, null);
        channel.queueBind(queue, exchange, routingKey);

        try {
            channel.close();// 在这里关闭channel是什么原因
        } catch (TimeoutException e) {
            e.printStackTrace();
            log.info("close channel time out ", e);
        }
    }

    private void buildTopic(String exchange, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        /*
            b表示：durable 持久化
            b1表示：autoDelete 自动删除
         */
        channel.exchangeDeclare(exchange, MQConstants.EXCHANGE_TOPIC_TYPE, true, false, null);
    }

//    private QueueingConsumer buildQueueConsumer(Connection connection, String queue) {
//        try {
//            Channel channel = connection.createChannel(false);
//            QueueingConsumer consumer = new QueueingConsumer(channel);
//
//            //通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。
//            //换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它
//            channel.basicQos(1);
//            channel.basicConsume(queue, false, consumer);
//
//            return consumer;
//        } catch (Exception e) {
//            e.printStackTrace();
//            log.info("build queue consumer error : ", e);
//
//            try {
//                Thread.sleep(MQConstants.ONE_SECOND);
//            } catch (InterruptedException inE) {
//                inE.printStackTrace();
//            }
//
//            return buildQueueConsumer(connection, queue);
//        }
//    }

    //for test
    public int getMessageCount(final String queue) throws IOException {
        Connection connection = connectionFactory.createConnection();
        final Channel channel = connection.createChannel(false);
        final AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queue);

        return declareOk.getMessageCount();
    }
}
